#include "wl_message_queue.h"

#include "wl_inner.h"
#include "wl_list.h"
#include "wl_malloc.h"
#include "wl_semaphore.h"
#include "wl_trace.h"

#ifdef __cplusplus
extern "C" {
#endif /* __cplusplus */

wl_int32 wl_message_queue_create(const wl_char*        name,
                                 OUT WL_MESSAGE_QUEUE* msg_queue,
                                 wl_uint32             msg_count,
                                 wl_uint32             msg_size) {
    if (msg_queue) {
        wl_char semname[QUEUE_NAME_MAX];
        sprintf(semname, "%s_locked", name);
        semname[QUEUE_NAME_MAX - 1] = '\0';
        if (0 != wl_mutex_create(semname, &msg_queue->locked)) {
            WLLOGW("wl_mutex_create %s fail", semname);
            return -1;
        }
        sprintf(semname, "%s_empty", name);
        semname[QUEUE_NAME_MAX - 1] = '\0';
        if (0 != wl_semaphore_create(semname, &msg_queue->empty, 0)) {
            WLLOGW("wl_semaphore_create %s fail", semname);
            wl_mutex_release(&msg_queue->locked);
            wl_mutex_delete(&msg_queue->locked);
            return -1;
        }
        msg_queue->limit = msg_count;
        wl_list_init(&msg_queue->queue);
        WLLOGI("queue %s create success", name);
        return 0;
    }
    WLLOGI("queue %s create fail", name);
    return -1;
}

wl_int32 wl_message_queue_send(WL_MESSAGE_QUEUE* msg_queue, void* msg, wl_size msg_size) {
    if (wl_null != msg_queue) {
        void*    element = wl_null;
        wl_int32 size    = 0;
        wl_mutex_acquire(&msg_queue->locked);
        size = wl_list_size(&msg_queue->queue);
        if (size >= msg_queue->limit) {
            wl_mutex_release(&msg_queue->locked);
            WLLOGT("%s msg_queue blocked size %d limit %d", msg_queue->name, size, msg_queue->limit);

            return -1;
        }
        element = wl_malloc(msg_size);
        if (wl_null == element) {
            wl_mutex_release(&msg_queue->locked);
            WLLOGW("%s malloc fail", msg_queue->name);
            return -1;
        }
        memcpy(element, msg, msg_size);
        if (0 > wl_list_add(&msg_queue->queue, element, -1)) {
            wl_free(element);
            wl_mutex_release(&msg_queue->locked);
            WLLOGW("%s wl_list_add fail", msg_queue->name);
            return -1;
        }
        wl_mutex_release(&msg_queue->locked);
        wl_semaphore_post(&msg_queue->empty);

        WLLOGT("%s send success", msg_queue->name);
        return 0;
    }
    return -1;
}

wl_int32 wl_message_queue_receive(WL_MESSAGE_QUEUE* msg_queue, void* msg, wl_size msg_size) {
    if (wl_null != msg_queue) {
        void* element = wl_null;
        if (0 != wl_semaphore_wait(&msg_queue->empty)) {
            WLLOGW("%s wl_semaphore_wait fail", msg_queue->name);
            return -1;
        }
        if (0 != wl_mutex_acquire(&msg_queue->locked)) {
            WLLOGW("%s wl_mutex_acquire fail", msg_queue->name);
            return -1;
        }
        element = wl_list_get(&msg_queue->queue, 0);
        wl_list_remove(&msg_queue->queue, 0);
        wl_mutex_release(&msg_queue->locked);
        if (element) {
            memcpy(msg, element, msg_size);
            wl_free(element);
            WLLOGT("%s recv success", msg_queue->name);
            return 0;
        }
        WLLOGT("%s recv fail", msg_queue->name);
        return -1;
    }
    return -1;
}

wl_int32 wl_message_queue_delete(WL_MESSAGE_QUEUE* msg_queue) {
    if (wl_null != msg_queue) {
        wl_semaphore_post(&msg_queue->empty);
        wl_mutex_release(&msg_queue->locked);
        wl_mutex_delete(&msg_queue->empty);
        wl_semaphore_delete(&msg_queue->locked);
        wl_list_free(&msg_queue->queue);
        return 0;
    }
    return -1;
}

#ifdef __cplusplus
}
#endif /* __cplusplus */
